package org.zkdemo.zkapi;

import java.io.Closeable;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.List;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zkdemo.api.capabilities.Capabilities;

public class ZkService implements Closeable, Watcher
{
    
    private static final Logger      LOGGER = LoggerFactory.getLogger(ZkService.class);
    
    private ZooKeeper                zooKeeper;
    
    private String                   host;
    private int                      port;
    private int                      sessionTimeout;
    
    private String                   rootName;
    private Collection<Capabilities> capabilities;
    
    private String                   capabilitiesString;
    
    public ZkService()
    {
        
    }
    
    public ZkService(String host, int port)
    {
        super();
        this.host = host;
        this.port = port;
    }
    
    public void init() throws IOException
    {
        LOGGER.info("Trying to connect to: " + host + ":" + port);
        
        zooKeeper = new ZooKeeper(host + ":" + port, sessionTimeout, this);
    }
    
    @Override
    public void close() throws IOException
    {
        try
        {
            zooKeeper.close();
        }
        catch (InterruptedException e)
        {
            throw new IOException("Error closing session.", e);
        }
    }
    
    public void setCapabilitiesString(String capabilitiesString)
    {
        this.capabilitiesString = capabilitiesString;
    }
    
    public Collection<Capabilities> getCapabilities()
    {
        if (capabilities == null)
        {
            capabilities = ConfigUtils.deserializeCapabilities(capabilitiesString);
        }
        
        return Collections.unmodifiableCollection(capabilities);
    }
    
    public String getCapabilitiesString()
    {
        return capabilitiesString;
    }
    
    public void setRootName(String rootName)
    {
        this.rootName = rootName;
    }
    
    public String getRootName()
    {
        return rootName;
    }
    
    public void setSessionTimeout(int sessionTimeout)
    {
        this.sessionTimeout = sessionTimeout;
    }
    
    public int getSessionTimeout()
    {
        return sessionTimeout;
    }
    
    public String getHost()
    {
        return host;
    }
    
    public void setHost(String host)
    {
        this.host = host;
    }
    
    public int getPort()
    {
        return port;
    }
    
    public void setPort(int port)
    {
        this.port = port;
    }
    
    public void registerModule(String id)
    {
        try
        {
            createRoot();
        }
        catch (KeeperException | InterruptedException e)
        {
            LOGGER.error("Error creating node.", e);
        }
        
        try
        {
            LOGGER.info("Trying to register this module id: " + id);
            zooKeeper.create("/" + rootName + "/modules/" + id, ConfigUtils.serializeCapabilities(getCapabilities()),
                             ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        }
        catch (NodeExistsException e)
        {
            LOGGER.warn("Module exists. Exiting ...");
            System.exit(-1);
        }
        catch (KeeperException | InterruptedException e)
        {
            LOGGER.error("Error creating node.", e);
            System.exit(-2);
        }
        
        LOGGER.info("Registered module id: " + id);
    }
    
    public void createRoot() throws KeeperException, InterruptedException
    {
        LOGGER.info("Trying to create root node: " + rootName);
        try
        {
            zooKeeper.create("/" + rootName, rootName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        catch (NodeExistsException e)
        {
            LOGGER.warn("/" + rootName + " " + "already exists. Ignoring ...");
        }
        
        try
        {
            zooKeeper.create("/" + rootName + "/" + ConfigUtils.MODULES, rootName.getBytes(),
                             ZooDefs.Ids.OPEN_ACL_UNSAFE,
                             CreateMode.PERSISTENT);
        }
        catch (NodeExistsException e)
        {
            LOGGER.warn("/" + rootName + "/modules already exists. Ignoring ...");
        }
        
        try
        {
            zooKeeper.create("/" + rootName + "/" + ConfigUtils.DISCOVERY, rootName.getBytes(),
                             ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        catch (NodeExistsException e)
        {
            LOGGER.warn("/" + rootName + "/discovery already exists. Ignoring ...");
        }
    }
    
    public void addAddressToDiscovery(String host) throws KeeperException, InterruptedException
    {
        try
        {
            zooKeeper.create("/" + rootName + "/" + ConfigUtils.DISCOVERY + "/" + host, rootName.getBytes(),
                             ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        catch (NodeExistsException e)
        {
            LOGGER.warn("/" + rootName + "/" + ConfigUtils.DISCOVERY + "/" + host + " already exists. Ignoring ...");
        }
    }
    
    public List<String> getDiscoverableItems(Watcher watcher) throws KeeperException, InterruptedException
    {
        return zooKeeper.getChildren("/" + rootName + "/" + ConfigUtils.DISCOVERY, watcher);
    }
    
    @Override
    public void process(WatchedEvent event)
    {
        LOGGER.info(String.valueOf(event));
        
        if (KeeperState.SyncConnected.equals(event.getState()))
        {
            // managementService.registerSelf();
        }
    }
    
}
